
#ifndef _TTG_PIPELINE_DATASTREAMNODE_
#define _TTG_PIPELINE_DATASTREAMNODE_

#include "PackerNode.h"

namespace ttg_internal
{
	namespace pipeline
	{
		//Processes data streams and provides events.
		//Also responds for loops and subsystems.
		template <class InType, class OutType,
				  class ProcessInType, class ProcessOutType>
		class DataStreamNode :public PackerNode<InType, OutType, ProcessInType, ProcessOutType>
		{
			private:
				ttg::pipeline::ContextMap<ttg::threads::FastCriticalSection> lockers;
						//lockers for startIteration()/finishIteration().
			private:
				//Calls onBegin() for this node and all bindings.
				inline void callOnBegin(ttg::pipeline::ContextID ID)
				{
					for (size_t i = 0; i < getBindings().size(); i++)
						getBindings()[i]->onBegin(ID);
					onBegin(ID);
				}
				//Calls onEnd() for this node and all bindings.
				inline void callOnEnd(ttg::pipeline::ContextID ID)
				{
					onEnd(ID);
					for (size_t i = 0; i < getBindings().size(); i++)
						getBindings()[i]->onEnd(ID);
				}
				//Checks, should iteration be finished or not. Calls onEnd().
				bool checkCompleteness(ttg::pipeline::ContextID ID)
				{
					NodeState &state = getState(ID);
					bool res = false;
					lockers[ID].enter();
					state.getLocker().enter();
					if (state.getActiveTaskCount() == 0 &&
						state.getActiveInputCount() == 0)
					{
						if (state.getProcessingState() != Ended)
						{
							state.getLocker().leave();
							state.getProcessingState() = Ended;
							callOnEnd(ID);
							getIContext().getTaskGatherer(ID).addTasks(packBuffers(ID));
							state.getLocker().enter();
							res = state.getActiveTaskCount() == 0;
						}
						else
							res = true;
					}
					if (res)
					{
						//Avast!
						getInfo().getLocker().enter();
						getInfo().getCoefficient() = getInfo().getTimeDifficulty() /
														std::max<double>(getInfo().getAbstractDifficulty(), 1.0);
						getInfo().getLocker().leave();
					}
					state.getLocker().leave();
					lockers[ID].leave();
					return res;
				}
			protected:
				//PackerNode member. Changes state of node after each performed task.
				virtual void taskHasBeenPerformed_CheckCompleteness(ttg::pipeline::ContextID ID)
				{
					if (checkCompleteness(ID))
						for (size_t i = 0; i < getOutputs().size(); i++)
							getOutputs()[i]->finishIteration(ID);
				}
				//Will be called before processing of any task data.
				virtual void onBegin(ttg::pipeline::ContextID ID)
				{ /*nothing*/ }
				//Will be called after processing of all task data.
				virtual void onEnd(ttg::pipeline::ContextID ID)
				{ /*nothing*/ }
				//Will be called after any modification of pipeline.
				virtual void onModification()
				{ /*nothing*/ }
			public:
				inline DataStreamNode(const char *name, bool is_serial)
					:PackerNode(name, is_serial)
				{ /*nothing*/ }
				//Safely calls onBegin()/preBegin() and sends startIteration to all output nodes.
				virtual void startIteration(ttg::pipeline::ContextID ID)
				{
					NodeState &state = getState(ID);
					lockers[ID].enter();
					state.getLocker().enter();
					if (state.getProcessingState() != Begined)
					{
						//Resetting informations.
						getInfo().getLocker().enter();
						getInfo().getAbstractDifficulty() = 0;
						getInfo().getTimeDifficulty() = 0.0;
#ifdef TTG_PL_VISUALIZER
						getInfo().getTaskCount() = 0;
						getInfo().getActiveTaskCount() = 0;
						for (size_t i = 0; i < getInfo().getThreadUsage().size(); i++)
							getInfo().getThreadUsage()[i] = 0.0;
#endif
						getInfo().getLocker().leave();
						//Calling onBegin().
						state.getProcessingState() = Begined;
						state.getLocker().leave();
						callOnBegin(ID);
						getIContext().getTaskGatherer(ID).addTasks(packBuffers(ID));
						state.getLocker().enter();
						state.getActiveInputCount() =  (int)getParametrizedInputs().size();
						for (size_t i = 0; i < getOutputs().size(); i++)
							getOutputs()[i]->startIteration(ID);
					}
					state.getLocker().leave();
					lockers[ID].leave();
				}
				//Safely calls onEnd() and sends finishIteration() to all output nodes.
				virtual void finishIteration(ttg::pipeline::ContextID ID)
				{
					getState(ID).getLocker().enter();
					getState(ID).getActiveInputCount() = max(0, getState(ID).getActiveInputCount() - 1);
					getState(ID).getLocker().leave();
					if (this->checkCompleteness(ID))
					{
						for (size_t i = 0; i < getOutputs().size(); i++)
							getOutputs()[i]->finishIteration(ID);
					}
				}
				//This method will be called after each pipeline modification.
				//NavigatorNode member. In case of error throws Error().
				virtual void onPipelineRecreation()
				{
					onModification();
				}
		};
	}
}

#endif
